package HandshakeJoin;

import org.apache.s4.core.ProcessingElement;
import org.apache.s4.core.Stream;
import org.apache.s4.core.RemoteStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RequestConverter extends ProcessingElement
{
    transient private static final Logger logger = LoggerFactory
            .getLogger(RequestConverter.class);

    private RemoteStream outputStream;
    private String joinId;

    @Override
    protected void onCreate() {}

    public void onTime() {
        QueryRequest req = constructOneTimeQueryByJoinKey(225);
        // QueryRequest req = constructOneTimeQueryByJoinKey(225, 235);

        outputStream.put(req);
        logger.info(req.toString());
    }

    @Override
    protected void onRemove() {}

    private QueryRequest constructOneTimeQueryByJoinKey(final int value) {
        QueryRequest req = new QueryRequest(getId(), joinId,
                QueryRequest.Type.ONE_TIME);
        req.setSubtype(QueryRequest.Subtype.EQUAL_BY_JOIN_KEY).setArgument(
                "value", Integer.toString(value));

        return req;
    }

    private QueryRequest constructOneTimeQueryByJoinKey(final int valueLower,
            final int valueUpper) {
        QueryRequest req = new QueryRequest(getId(), joinId,
                QueryRequest.Type.ONE_TIME);
        req.setSubtype(QueryRequest.Subtype.RANGE_BY_JOIN_KEY)
                .setArgument("valueLower", Integer.toString(valueLower))
                .setArgument("valueUpper", Integer.toString(valueUpper));

        return req;
    }

    private QueryRequest constructOneTimeQueryByTimestamp(final long ts) {
        QueryRequest req = new QueryRequest(getId(), joinId,
                QueryRequest.Type.ONE_TIME);
        req.setSubtype(QueryRequest.Subtype.EQUAL_BY_TIMESTAMP).setArgument(
                "timestamp", Long.toString(ts));

        return req;
    }

    private QueryRequest constructOneTimeQueryByTimestamp(final long tsLower,
            final long tsUpper) {
        QueryRequest req = new QueryRequest(getId(), joinId,
                QueryRequest.Type.ONE_TIME);
        req.setSubtype(QueryRequest.Subtype.RANGE_BY_TIMESTAMP)
                .setArgument("timestampLower", Long.toString(tsLower))
                .setArgument("timestampUpper", Long.toString(tsUpper));

        return req;
    }

    public RequestConverter setJoinId(final String joinId) {
        this.joinId = joinId;

        return this;
    }

    public RequestConverter setOutputStream(final RemoteStream outputStream) {
        this.outputStream = outputStream;

        return this;
    }
}